
#include "../utils/BitVector.h"
#include "../utils/Hash.h"
#include "Broker5Client.h"
#include "BlockWriteServer.h"
#include "BlockWriteClient.h"

BlockWriteClient::BlockWriteClient(int inoofthreads, KeyDB *itkeydb, apr_pool_t *upperpool)
{
   tkeydb = itkeydb;
   apr_status_t status;
   status = apr_pool_create(&mp, upperpool);
   noofthreads = inoofthreads;
   allblocksstop = 0;
   watchstop = 0;
   uploadarray = (ServerPipeDatagrams *)apr_palloc(mp, sizeof(ServerPipeDatagrams)*noofthreads);
   for (int i = 0; i < noofthreads; i++)
   {
      uploadarray[i].localmp = NULL;
      uploadarray[i].active = 0;
      uploadarray[i].blockID = 0;
      uploadarray[i].udp_sender = NULL;
      uploadarray[i].udp_hostname = NULL;
      uploadarray[i].udp_portno = 0;
      uploadarray[i].tcp_hostname = NULL;
      uploadarray[i].tcp_portno = 0;
      void *mem_for_new_sessionstart_lock = apr_palloc(mp, sizeof(APRMutex));
      uploadarray[i].sessionstart = new(mem_for_new_sessionstart_lock) APRMutex(mp);
   }
   blocks_threads = (apr_thread_t **)apr_palloc(mp, sizeof(apr_thread_t *)*noofthreads);
   blocks_threadattrs = (apr_threadattr_t **)apr_palloc(mp, sizeof(apr_threadattr_t *)*noofthreads);
}

/** Start handing blocks to the server (the block threads run this).
 * 
 * @param threadid The integer thread ID number.
 * 
 * @param localmp The memory pool local to this thread. */
void BlockWriteClient::handleBlocksOnThread(int threadid, apr_pool_t *localmp)
{
   uint8_t packetdata[sizeof(uint32_t) + PACKET_BLOCK_SIZE];
   uploadarray[threadid].sessionstart->lock();
   while (!allblocksstop)
   {
      uploadarray[threadid].sessionstart->wait();
      uploadarray[threadid].udp_sender = new UDPSend(uploadarray[threadid].udp_hostname, uploadarray[threadid].udp_portno, localmp);
      while (!(uploadarray[threadid].server_says_complete))
      {
         // Initialise bit vector manipulation. Send all the blocks
         // with zero bit vector associated with them.
         for (int i = 0; i < (ACTIVE_BLOCK_SIZE / PACKET_BLOCK_SIZE); i++)
            if (BitVector::getBitFast(uploadarray[threadid].filled_bit_vector, i) == 0)
            {
               // Build UDP packet
               ((uint16_t *)packetdata)[0] = uploadarray[threadid].blockID;
               ((uint16_t *)packetdata)[1] = i;
               memcpy(&(((uint16_t *)packetdata)[2]), &(((uint8_t *)(uploadarray[threadid].active_block_data))[i*PACKET_BLOCK_SIZE]), PACKET_BLOCK_SIZE);
               uploadarray[threadid].udp_sender->sendPacket(packetdata, sizeof(uint32_t) + PACKET_BLOCK_SIZE);
            }
         Broker5Client *b5client = new Broker5Client(uploadarray[threadid].tcp_hostname, uploadarray[threadid].tcp_portno, localmp);
         BrokerMessage *bmsg = new BrokerMessage(localmp);
         bmsg->setSender(uploadarray[threadid].sender);
         bmsg->setRecipient(uploadarray[threadid].recipient);
         bmsg->addTypeHeaderNOCRLF("TRANSFER_REQUEST_STATUS", uploadarray[threadid].recipient);
         bmsg->addHeaderLineNOCRLF("BlockID: %d", uploadarray[threadid].blockID);
         b5client->exchangeMessage(*bmsg, *tkeydb);
         if (bmsg->getType())
         {
            // The server has forgotten us! - we die
            if (strcmp(bmsg->getType(), "TRANSFER_BLOCK_UNKNOWN") == 0)
               uploadarray[threadid].server_says_complete = 1;
            // The payload is bit vector, copy it into the active
            // area.
            if (strcmp(bmsg->getType(), "TRANSFER_PROGRESS") == 0)
               if (bmsg->getPayloadLength(0) == ((ACTIVE_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8)))
                  memcpy(uploadarray[threadid].filled_bit_vector, bmsg->getPayload(0), bmsg->getPayloadLength(0));
            // Upload complete
            if (strcmp(bmsg->getType(), "TRANSFER_COMPLETE") == 0)
               uploadarray[threadid].server_says_complete = 1;
         }
      }
      delete uploadarray[threadid].udp_sender;
      uploadarray[threadid].udp_sender = NULL;
      uploadarray[threadid].blockID = 0;
      uploadarray[threadid].active = 0;
   }
   uploadarray[threadid].sessionstart->unlock();
}

/** Start handing blocks to the server (the block threads run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *BlockWriteClient::handleBlocksOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   BlockWriteClient *kthis = *((BlockWriteClient **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&mp, upperpool);
   CheckAPRError(status);

   kthis->handleBlocksOnThread(i, mp);

   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Start watching over the active blocks (this is a separate thread).
 * 
 * @param localmp The local memory pool to this thread that is
 * destroyed afterwards. */
void BlockWriteClient::watchFromThread(apr_pool_t *localmp)
{
}

/** Start watching over the active blocks (this is a separate thread).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to the this pointer.
 * 
 * @return Return is NULL. */
void *BlockWriteClient::watchFromThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   BlockWriteServer *kthis = (BlockWriteServer *)args;
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   apr_pool_create(&mp, upperpool);
   
   kthis->watchFromThread(mp);

   ERR_remove_state(0);
   apr_pool_destroy(mp);
   status = apr_thread_exit(tthread, APR_SUCCESS);
   CheckAPRError(status);
   return NULL;
}

/** Put data to upload onto the queue.
 * 
 * @param hostname Host to upload the block to.
 * 
 * @param portno Port to upload the block to.
 * 
 * @param suggested_filename Suggested filename for the file that
 * contains the block.
 * 
 * @param offset Offset that the block has inside the file.
 * 
 * @param filesize File size of the completed file.
 * 
 * @param data_pointer Pointer to the data to upload.
 * 
 * @param blocklength The length of the block of data to be sent.
 * 
 * @param unencrypted_hash The hash of the unencrypted data.
 * 
 * @return Zero on successfully queueing the block, non-zero on
 * failure. */
int BlockWriteClient::queueBlock(const char *hostname, int portno,
   const char *suggested_filename, apr_off_t offset, apr_size_t filesize, uint8_t *data_pointer,
   apr_size_t blocklength, const char *unencrypted_hash)
{
   if (hostname == NULL)
   {
      errprint("can't queue block to transfer to NULL hostname");
      return 1;
   }
   if (suggested_filename == NULL)
   {
      errprint("can't queue the block with no suggested filename");
      return 1;
   }
   if (unencrypted_hash == NULL)
   {
      errprint("can't match unencrypted block hashes");
      return 1;
   }
   BlockForUpload *bfupload = (BlockForUpload *)malloc(sizeof(BlockForUpload));
   bfupload->udp_hostname = strdup(hostname);
   bfupload->udp_portno = portno;
   apr_size_t newblocksize = ((blocklength / HASHING_BLOCK_SIZE) + 1) * HASHING_BLOCK_SIZE;
   // Allocate overflowing array
   bfupload->active_block_data = (uint8_t *)malloc(sizeof(uint8_t)*newblocksize);
   // Zero overflowing area.
   memset(&(bfupload->active_block_data[newblocksize - HASHING_BLOCK_SIZE]), 0, HASHING_BLOCK_SIZE);
   // Copy the pointed to data.
   memcpy(bfupload->active_block_data, data_pointer, blocklength);
   bfupload->active_block_hashes = NULL;
   bfupload->unencrypted_hash = strdup(unencrypted_hash);
   bfupload->filename = strdup(suggested_filename);
   bfupload->offset = offset;
   bfupload->filelength = filesize;
   bfupload->blocklength = blocklength;
   queue_mutex->lock();
   uploadqueue.push_back(bfupload);
   queue_mutex->signal();
   queue_mutex->unlock();
}

/** Hash the active data block of the local thread. This must be done
 * before anything else, but it should also be on a single upload
 * thread to not block things.
 * 
 * @param threadid The thread (and #uploadarray) ID and index.
 * 
 * @param localmp The local memory pool to this thread and function.
 * 
 * @return One if this slot is to be returned to a free state. Zero if
 * this slot is to be actively behaving as a client. */
int BlockWriteClient::requestUploadSlot(int threadid, apr_pool_t *localmp)
{
   queue_mutex->lock();
   queue_mutex->wait();
   if (uploadqueue.empty()) return 1;
   BlockForUpload *mbfupload = uploadqueue.front();
   int required_packet_blocks = ((mbfupload->blocklength + PACKET_BLOCK_SIZE - 1) / PACKET_BLOCK_SIZE);
   int required_hashing_blocks = ((mbfupload->blocklength + HASHING_BLOCK_SIZE - 1) / HASHING_BLOCK_SIZE);
   if (mbfupload->active_block_hashes == NULL)
   {
      uint8_t *hash_data = 
         (uint8_t *)malloc(sizeof(uint8_t)*(SHA1_DIGEST_LENGTH * required_hashing_blocks));
      for (int i = 0; i < required_hashing_blocks; i++)
         Hash::getBinarySHA1Hash((const char *)(&(mbfupload->active_block_data[i*HASHING_BLOCK_SIZE])), HASHING_BLOCK_SIZE,
            &(hash_data[i*SHA1_DIGEST_LENGTH]));
      mbfupload->active_block_hashes = hash_data;
   }
   if (uploadarray[threadid].udp_hostname) free(uploadarray[threadid].udp_hostname);
   uploadarray[threadid].udp_hostname = strdup(mbfupload->udp_hostname);
   uploadarray[threadid].udp_portno = mbfupload->udp_portno;
   if (uploadarray[threadid].tcp_hostname) free(uploadarray[threadid].tcp_hostname);
   uploadarray[threadid].tcp_hostname = strdup(mbfupload->tcp_hostname);
   uploadarray[threadid].tcp_portno = mbfupload->tcp_portno;
   uploadarray[threadid].active_block_data = mbfupload->active_block_data;
   if (uploadarray[threadid].sender) free(uploadarray[threadid].sender);
   uploadarray[threadid].sender = strdup(mbfupload->sender);
   if (uploadarray[threadid].recipient) free(uploadarray[threadid].recipient);
   uploadarray[threadid].recipient = strdup(mbfupload->recipient);
   for (int i = 0; i < ((ACTIVE_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8)); i++)
      uploadarray[threadid].filled_bit_vector[i] = 0;
   
   Broker5Client *b5client = new Broker5Client(uploadarray[threadid].tcp_hostname,
      uploadarray[threadid].tcp_portno, localmp);
   BrokerMessage *bmsg = new BrokerMessage(localmp);
   bmsg->setSender(uploadarray[threadid].sender);
   bmsg->setRecipient(uploadarray[threadid].recipient);
   bmsg->addTypeHeaderNOCRLF("TRANSFER_START_BLOCK", uploadarray[threadid].recipient);
   bmsg->addHeaderLineNOCRLF("Transfer-Name: %s", mbfupload->filename);
   bmsg->addHeaderLineNOCRLF("Transfer-Length: %" APR_SIZE_T_FMT, mbfupload->filelength);
   bmsg->addHeaderLineNOCRLF("Transfer-Offset: %" APR_OFF_T_FMT, mbfupload->offset);
   bmsg->addHeaderLineNOCRLF("PlainText-Hash: %s", mbfupload->unencrypted_hash);
   bmsg->addPayloadBuffer((const char *)(mbfupload->active_block_hashes), SHA1_DIGEST_LENGTH * required_hashing_blocks);
   b5client->exchangeMessage(*bmsg, *tkeydb);
   int uploading_okay = 0;
   int uploading_complete = 0;
   if (bmsg->getType())
   {
      // Transfer ready to start, server ready.
      if (strcmp(bmsg->getType(), "TRANSFER_START_AUTHORISED") == 0)
         uploading_okay = 1;
      if (uploading_okay)
      {
         // This is where the bit vector is initialised!
         apr_size_t bv_length_bytes = bmsg->getPayloadLength(0);
         memcpy(uploadarray[threadid].filled_bit_vector, bmsg->getPayload(0), bv_length_bytes);
      }
      // Transfer completed before, don't upload but remove this from
      // the queue.
      if (strcmp(bmsg->getType(), "TRANSFER_COMPLETE") == 0)
         { uploading_okay = 1; uploading_complete = 1; }
   }
   delete bmsg;
   delete b5client;
   // uploading_complete should also set uploading_okay
   if (uploading_okay)
   {
      uploadqueue.pop_front();
      free(mbfupload->udp_hostname);
      free(mbfupload->tcp_hostname);
      free(mbfupload->active_block_hashes);
      free(mbfupload);
   }
   queue_mutex->unlock();
   if (uploading_complete) return 1;
   return 0;
}
